package cn.itcast.flink.transformation;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author lilulu
 */
/*
对数据流进行分割，使用sideOutput侧边输出算子实现，将奇数数字放在一个流，将偶数数字放在一个
流，todo：原来数据流中数据平方处理
*/
// step1、定义分割流标签
// step2、调用process函数，对流中数据处理及打标签
// step3、判断数据是奇数还是偶数，打上对应标签
// step4、获取侧边流，依据标签
public class TransformationSideOutputsDemo {
    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 数据源-source
        DataStreamSource<Integer> inputStream = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        // 3. 数据转换-transformation
        OutputTag<Integer> oddTag = new OutputTag<Integer>("side-odd") {};
        OutputTag<Integer> evenTag = new OutputTag<Integer>("side-even") {};
        SingleOutputStreamOperator<String> mainStream = inputStream.process(
                new ProcessFunction<Integer, String>() {
                    @Override
                    public void processElement(Integer value, Context context, Collector<String> collector) throws Exception {
//                        判断数据是奇数还是偶数，打上对应标签
                        if (value % 2 == 0) {
                            context.output(evenTag, value);
                        } else {
                            context.output(oddTag, value);
                        }
                    }
                }
        );
        // 4. 数据终端-sink
        mainStream.printToErr();
        DataStream<Integer> sideOutput = mainStream.getSideOutput(oddTag);
        sideOutput.print("odd>");
        DataStream<Integer> sideOutput1 = mainStream.getSideOutput(evenTag);
        sideOutput1.print("even>");
        // 5. 触发执行-execute
        env.execute("TransformationSideOutputsDemo");
    }
}